package student.tangllei

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Sheep.Old @ 64341393
  * Created 2018/3/28
  */
object Tldemo {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
          .setMaster("local[*]")
          .setAppName(Tldemo.getClass.getSimpleName) .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
          .set("spark.sql.parquet.compression.codec", "snappy")

        val sc = new SparkContext(conf)

        val sql = new SQLContext(sc)

        val file = sc.textFile("2016-10-01_06_p1_invalid.1475274123982.log.FINISH")
        val caseFile= file.map(_.split(",")).filter(arr=>{

            arr.length>=85
        })
        .map(arr => Row(
            arr(0),
            arr(1).toInt,
            arr(2).toInt,
            arr(3).toInt,
            arr(4).toInt,
            arr(5),
            arr(6),
            arr(7).toInt,
            arr(8).toInt,
            arr(9).toDouble,
            arr(10).toDouble,
            arr(11),
            arr(12),
            arr(13),
            arr(14),
            arr(15),
            arr(16),
            arr(17).toInt,
            arr(18),
            arr(19),
            arr(20),
            arr(21),
            arr(22),
            arr(23),
            arr(24),
            arr(25),
            arr(26).toInt,
            arr(27),
            arr(28).toInt,
            arr(29),
            arr(30).toInt,
            arr(31).toInt,
            arr(32).toInt,
            arr(33),
            arr(34).toInt,
            arr(35).toInt,
            arr(36).toInt,
            arr(37),
            arr(38).toInt,
            arr(39).toInt,
            arr(40).toDouble,
            arr(41).toDouble,
            arr(42).toInt,
            arr(43),
            arr(44).toDouble,
            arr(45).toDouble,
            arr(46),
            arr(47),
            arr(48),
            arr(49),
            arr(50),
            arr(51),
            arr(52),
            arr(53),
            arr(54),
            arr(55),
            arr(56),
            arr(57).toInt,
            arr(58).toDouble,
            arr(59).toInt,
            arr(60).toInt,
            arr(61),
            arr(62),
            arr(63),
            arr(64),
            arr(65),
            arr(66),
            arr(67),
            arr(68),
            arr(69),
            arr(70),
            arr(71),
            arr(72),
            arr(73).toInt,
            arr(74).toDouble,
            arr(75).toDouble,
            arr(76).toDouble,
            arr(77).toDouble,
            arr(78).toDouble,
            arr(79),
            arr(80),
            arr(81),
            arr(82),
            arr(83),
            arr(84).toInt))
        val schema = StructType(Array(
            StructField("sessionid",StringType),
            StructField("advertisersid",IntegerType),
            StructField("adorderid",IntegerType),
            StructField("adcreativeid",IntegerType),
            StructField("adplatformproviderid",IntegerType),
            StructField("sdkversion",StringType),
            StructField("adplatformkey",StringType),
            StructField("putinmodeltype",IntegerType),
            StructField("requestmode",IntegerType),
            StructField("adprice",DoubleType),
            StructField("adppprice",DoubleType),
            StructField("requestdate",StringType),
            StructField("ip",StringType),
            StructField("appid",StringType),
            StructField("appname",StringType),
            StructField("uuid",StringType),
            StructField("device",StringType),
            StructField("client",IntegerType),
            StructField("osversion",StringType),
            StructField("density",StringType),
            StructField("pw",StringType),
            StructField("ph",StringType),
            StructField("long",StringType),
            StructField("lat",StringType),
            StructField("provincename",StringType),
            StructField("cityname",StringType),
            StructField("ispid",IntegerType),
            StructField("ispname",StringType),
            StructField("networkmannerid",IntegerType),
            StructField("networkmannername",StringType),
            StructField("iseffective",IntegerType),
            StructField("isbilling",IntegerType),
            StructField("adspacetype",IntegerType),
            StructField("adspacetypename",StringType),
            StructField("devicetype",IntegerType),
            StructField("processnode",IntegerType),
            StructField("apptype",IntegerType),
            StructField("district",StringType),
            StructField("paymode",IntegerType),
            StructField("isbid",IntegerType),
            StructField("bidprice",DoubleType),
            StructField("winprice",DoubleType),
            StructField("iswin",IntegerType),
            StructField("cur",StringType),
            StructField("rate",DoubleType),
            StructField("cnywinprice",DoubleType),
            StructField("imei",StringType),
            StructField("mac",StringType),
            StructField("idfa",StringType),
            StructField("openudid",StringType),
            StructField("androidid",StringType),
            StructField("rtbprovince",StringType),
            StructField("rtbcity",StringType),
            StructField("rtbdistrict",StringType),
            StructField("rtbstreet",StringType),
            StructField("storeurl",StringType),
            StructField("realip",StringType),
            StructField("isqualityapp",IntegerType),
            StructField("bidfloor",DoubleType),
            StructField("aw",IntegerType),
            StructField("ah",IntegerType),
            StructField("imeimd5",StringType),
            StructField("macmd5",StringType),
            StructField("idfamd5",StringType),
            StructField("openudidmd5",StringType),
            StructField("androididmd5",StringType),
            StructField("imeisha1",StringType),
            StructField("macsha1",StringType),
            StructField("idfasha1",StringType),
            StructField("openudidsha1",StringType),
            StructField("androididsha1",StringType),
            StructField("uuidunknow",StringType),
            StructField("userid",StringType),
            StructField("iptype",IntegerType),
            StructField("initbidprice",DoubleType),
            StructField("adpayment",DoubleType),
            StructField("agentrate",DoubleType),
            StructField("lomarkrate",DoubleType),
            StructField("adxrate",DoubleType),
            StructField("title",StringType),
            StructField("keywords",StringType),
            StructField("tagid",StringType),
            StructField("callbackdate",StringType),
            StructField("channelid",StringType),
            StructField("mediatype",IntegerType)
        ))


        // 第二种方式获取dataframe  自定义schema信息 + RDD[Row]
        val pdf = sql.createDataFrame(caseFile,schema)

        //pdf.printSchema()
        //println(pdf.schema)

        //pdf.write.parquet("./bbb")

        val parquet = sql.read.parquet("./bbb")
         parquet.printSchema()
        parquet.registerTempTable("sssss")
        sql.sql("select  * from sssss").show()
        sc.stop()
    }
}
